package com.bubbles.workspace.sync.service.doris;


import cn.hutool.v7.core.date.DateTime;
import cn.hutool.v7.core.date.DateUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.bubbles.workspace.sync.entity.source.metadata.*;
import com.bubbles.workspace.sync.enums.SourceEnabledReason;
import com.bubbles.workspace.sync.key.source.metadata.SourceColumnKey;
import com.bubbles.workspace.sync.key.source.metadata.SourceTableKey;
import com.bubbles.workspace.sync.properties.SourceProperties;
import com.bubbles.workspace.sync.properties.SshProperties;
import com.bubbles.workspace.sync.service.source.constraint.SourceConstraintForeignService;
import com.bubbles.workspace.sync.service.source.constraint.SourceConstraintPrimaryService;
import com.bubbles.workspace.sync.service.source.constraint.SourceConstraintUniqueService;
import com.bubbles.workspace.sync.service.source.constraint.SourceIndexUniqueService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDimensionService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeColumnService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeTableService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceIncrementService;
import com.bubbles.workspace.sync.service.source.metadata.*;
import com.bubbles.workspace.sync.util.SshManager;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * <p>description: 源数据源-检查约束 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-20
 */
@DS("pg")
@Slf4j
@Service
public class DorisAnalyzeService {
    
    @Resource
    private SourceJdbcService sourceJdbcService;
    
    @Resource
    private SourceServerService sourceServerService;
    
    @Resource
    private SourceCatalogService sourceCatalogService;
    
    @Resource
    private SourceDatabaseService sourceDatabaseService;
    
    @Resource
    private SourceTableService sourceTableService;
    
    @Resource
    private SourceColumnService sourceColumnService;
    
    @Resource
    private SourceConstraintPrimaryService sourceConstraintPrimaryService;
    
    @Resource
    private SourceConstraintUniqueService sourceConstraintUniqueService;
    
    @Resource
    private SourceConstraintForeignService sourceConstraintForeignService;
    
    @Resource
    private DorisAnalyzeService sourceConstraintCheckService;
    
    @Resource
    private SourceIndexUniqueService sourceIndexUniqueService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeTableService sourceEnhanceDorisAnalyzeTableService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeColumnService sourceEnhanceDorisAnalyzeColumnService;
    
    @Resource
    private SourceEnhanceIncrementService sourceEnhanceIncrementService;
    
    @Resource
    private SourceEnhanceDimensionService sourceEnhanceDimensionService;
    
    @Resource
    private DorisAnalyzeService dorisAnalyzeService;
    
    @Resource
    private SshProperties sshProperties;
    
    @Resource
    private SourceProperties sourceProperties;
    
    @Resource
    private SshManager sshManager;
    
    @DS("doris")
    @DSTransactional
    public void dorisStatistic(SourceCatalog sourceCatalog, SourceDatabase sourceDatabase) {
        String catalogName = sourceCatalog.getSourceCatalogName();
        String databaseName = sourceDatabase.getSourceDatabaseName();
        SourceJdbc sourceJdbc = sourceCatalog.getSourceJdbc();
        String transformSystemCode = sourceCatalog.getTransform().getTransformSystemCode();
        SourceDatabase.Doris databaseDoris = sourceDatabase.getDoris();
        
        if (!sourceProperties.getDorisAnalyze().getEnable()) {
            log.info("[SYNC] {} -> {} -> Doris统计 -> [取消] 配置项 source.doris-analyze = false", catalogName, databaseName);
            return;
        }
        if (StringUtils.isEmpty(transformSystemCode)) {
            log.info("[SYNC] {} -> {} -> Doris统计 -> [取消] Doris Catalog 未设置名称", catalogName, databaseName);
            return;
        }
        if (!sourceCatalog.getDoris().getDorisConnected()) {
            log.info("[SYNC] {} -> {} -> Doris统计 -> [取消] Doris Catalog 无法连接", catalogName, databaseName);
            return;
        }
        
        // 每天分析一次
        DateTime intervalDay = DateUtil.endOfDay(DateUtil.offsetDay(DateUtil.today(), -sourceProperties.getExpireIntervalDay() - 1), false);
        Date dorisAnalyzeTime = databaseDoris.getDorisAnalyzeDatetime();
        if (dorisAnalyzeTime != null && intervalDay.before(dorisAnalyzeTime)) {
            log.info("[SYNC] {} -> {} -> Doris统计 -> [取消] Doris Catalog 上次分析时间 => {}", catalogName, databaseName, dorisAnalyzeTime);
            return;
        }
        
        // 记录 ANALYZE DATABASE 过程中出错的表
        Set<String> errorTables = new HashSet<>();
        // todo 逻辑转换为 Map<String, SourceTable> 以优化逻辑和入参
        // 记录 ANALYZE DATABASE 过程中待分析的表名, 将通过remove的方式逐个移除
        Set<String> todoAnalyzeTables = sourceTableService.listByDatabase(sourceDatabase)
                                                          .stream()
                                                          .map(SourceTable::getSourceTableName)
                                                          .collect(Collectors.toSet());
        Map<String, SourceTable> todoAnalyzeTables = sourceTableService.listByDatabase(sourceDatabase)
                                                                       .stream()
                                                                       .collect(Collectors.toMap(SourceTable::getSourceTableName, Function.identity()));
        // todo 逻辑转换为 Map<String, LIST<SourceColumn>> 以优化逻辑和入参
        // 记录 ANALYZE DATABASE 过程中待分析的列 [key:tableName,value:columnNames]
        Map<String, List<String>> todoAnalyzeColumnMap = sourceColumnService.listByDatabase(sourceDatabase)
                                                                            .stream()
                                                                            .collect(Collectors.groupingBy(SourceColumn::getSourceTableName, Collectors.mapping(SourceColumn::getSourceColumnName, Collectors.toList())));
        
        // 经过前版代码的试错, 如需实现该功能有以下坑需处理
        // 1.ANALYZE DATABASE           (异步) 表分析出错时,该表的任务状态会变为失败
        // 2.ANALYZE DATABASE WITH SYNC (同步) 表分析出错时,抛出异常,后续表达任务全部都会停止 => 所以只能采用异步的方式
        // 3.ANALYZE TABLE     WITH SYNC 表分析出错时,抛出异常.该过程非常耗时
        // 4.ANALYZE TABLE(ID) WITH SYNC 列分析出错时,抛出异常.该过程非常耗时.且在使用过程中出现了JdbcTemplate超时的问题
        
        // 综上所述,合理的方案是:
        // 1.清空统计任务
        //     1.1.查看并清空统计信息收集作业 SHOW ANALYZE
        //     1.2.停止正在后台执行的统计信息收集作业 KILL ANALYZE
        //     1.3.删除指定的统计信息收集作业的历史记录 DROP ANALYZE
        // 2.清空统计任务,重复执行防止之前执行的过程中任务存在状态变更
        // 3.异步分析 ANALYZE DATABASE
        // 4.无限循环处理 SHOW ANALYZE (此时只有表任务), 直到没有相关任务
        //     4.1.等待,优先等待是为了让Doris能够完整的生成任务列表
        //     4.2.查看统计信息收集作业 SHOW ANALYZE
        //     4.3.对于 STATE = 'RUNNING' 和 'PENDING' -> 继续循环标志为true
        //     4.4.对于 STATE = 'FAILED'   -> DROP ANALYZE JOB -> 记录失败的表
        //     4.5.对于 STATE = 'FINISHED' -> DROP ANALYZE JOB -> SHOW TABLE STATS 提取表统计数据 -> SHOW COLUMN STATS 提取列统计数据
        //   循环结束
        // 5.异步分析未分析的表(此过程应当不存在,是为了避免Doris对任务提取的不全才做处理) ANALYZE TABLE
        //     5.1.分析未分析的表
        //     5.2.重复执行4
        // 6.所有失败的表的所有列调用 ANALYZE TABLE(单列), 注意:此过程将非常耗时
        // 7.无限循环处理 SHOW ANALYZE (此时只有列任务), 直到没有相关任务
        //     7.1.等待,优先等待是为了让Doris能够完整的生成任务列表
        //     7.2.查看统计信息收集作业 SHOW ANALYZE
        //     7.3.对于 STATE = 'RUNNING' 和 'PENDING' -> 继续循环标志为true
        //     7.4.对于 STATE = 'FAILED'   -> DROP ANALYZE JOB -> 记录失败的[表:列]Map (手动分析的时候可能会直接用到)
        //     7.5.对于 STATE = 'FINISHED' -> -> DROP ANALYZE JOB -> SHOW COLUMN STATS 提取列统计数据
        //   结束循环
        // 8.返回 失败的[表:列]Map
        
        // 1.清空统计任务
        log.info("[SYNC] {} -> {} -> Doris统计 -> 清空任务表", catalogName, databaseName);
        dorisAnalyzeService.clearAnalyzeJob(sourceCatalog, sourceDatabase);
        // 2.清空统计任务,重复执行防止之前执行的过程中任务存在状态变更
        dorisAnalyzeService.clearAnalyzeJob(sourceCatalog, sourceDatabase);
        
        // 3.异步分析 ANALYZE DATABASE
        log.info("[SYNC] {} -> {} -> Doris统计 -> 异步分析表", catalogName, databaseName);
        dorisAnalyzeService.analyzeDatabase(sourceCatalog, sourceDatabase);
        
        // 4.无限循环处理 SHOW ANALYZE (此时只有表任务), 直到没有相关任务
        log.info("[SYNC] {} -> {} -> Doris统计 -> 处理表信息", catalogName, databaseName);
        dorisAnalyzeService.dealTableAsyncTask(sourceCatalog, sourceDatabase, errorTables, todoAnalyzeTables, todoAnalyzeColumnMap);
        
        // 5.异步分析未分析的表(是为了避免Doris对任务提取的不全才做处理) ANALYZE TABLE
        // 例如默认 SET GLOBAL auto_analyze_table_width_threshold = 300; 宽度超过300列的表不会进行自动分析,这里需要手动执行
        if (!todoAnalyzeTables.isEmpty()) {
            for (String tableName : todoAnalyzeTables) {
                log.info("[SYNC] {} -> {} -> Doris统计 -> 单独分析表 => {}", catalogName, databaseName, tableName);
                // 5.1.分析未分析的表
                dorisAnalyzeService.analyzeTable(sourceCatalog, sourceDatabase, tableName);
            }
            // 5.2.重复执行4
            dorisAnalyzeService.dealTableAsyncTask(sourceCatalog, sourceDatabase, errorTables, todoAnalyzeTables, todoAnalyzeColumnMap);
            if (!todoAnalyzeTables.isEmpty()) {
                log.error("[SYNC] {} -> {} -> Doris统计 -> 单独分析表 -> 未分析表 => {}", catalogName, databaseName, String.join(",", todoAnalyzeTables));
            }
        }
        
        // 6.单独调用 ANALYZE TABLE(单列), 注意:此过程将非常耗时
        // 注意此时 todoAnalyzeTables 中分析出错的表都被移出并放入了 errorTables. todoAnalyzeColumnMap 中的表没有变化.
        todoAnalyzeColumnMap.forEach((tableName, columnNames) -> {
            log.info("[SYNC] {} -> {} -> Doris统计 -> 单独分析列 => {}", catalogName, databaseName, tableName);
            // 分析失败的表的所有列
            if (errorTables.contains(tableName)) {
                columnNames.forEach(columnName -> {
                    log.info("[SYNC] {} -> {} -> Doris统计 -> 单独分析列 => {}.{}", catalogName, databaseName, String.format("%" + tableName.length() + "s", tableName), columnName);
                    dorisAnalyzeService.analyzeColumn(sourceCatalog, sourceDatabase, tableName, columnName);
                });
            } else {
                // 未分析的所有列 这里面包括 errorTables + todoAnalyzeTables 的表剩余列数据
                if (sourceProperties.getDorisAnalyze().getAllColumn()) {
                    columnNames.forEach(columnName -> {
                        log.info("[SYNC] {} -> {} -> Doris统计 -> 强制分析列 => {}.{}", catalogName, databaseName, tableName, columnName);
                        dorisAnalyzeService.analyzeColumn(sourceCatalog, sourceDatabase, tableName, columnName);
                    });
                }
            }
        });
        
        // 7.无限循环处理 SHOW ANALYZE (此时只有列任务), 直到没有相关任务
        log.info("[SYNC] {} -> {} -> Doris统计 -> 处理列信息", catalogName, databaseName);
        dorisAnalyzeService.dealColumnAsyncTask(sourceCatalog, sourceDatabase, todoAnalyzeColumnMap);
        
        // 8.记录库分析时间
        databaseDoris.setDorisAnalyzeDatetime(new Date());
        sourceDatabaseService.save(sourceDatabase);
        
        // 9.无法分析的表都在 todoAnalyzeTables 和 errorTables 中, 设置以便后续的分析任务
        for (String tableName : todoAnalyzeTables) {
            SourceTable sourceTable = sourceTableService.getById(new SourceTableKey(catalogName, databaseName, tableName));
            // 9.1.手动开启或关闭的表 -> 跳过
            if (SourceEnabledReason.MANUAL.equals(sourceTable.getSourceTableEnabledReason())) {
                return;
            }
            // 10.2.还原为初始采集状态
            sourceTable.setSourceTableEnabled(true);
            sourceTable.setSourceTableEnabledReason(SourceEnabledReason.AUTO);
            sourceTableService.save(sourceTable);
        }
        
        // 10.无法分析的列都在 todoAnalyzeColumnMap 中, 设置以便后续的分析任务
        todoAnalyzeColumnMap.forEach((tableName, columnNames) -> {
            columnNames.forEach(columnName -> {
                SourceColumn sourceColumn = sourceColumnService.getById(new SourceColumnKey(catalogName, databaseName, tableName, columnName));
                // 10.1.手动开启或关闭的列 -> 跳过
                if (SourceEnabledReason.MANUAL.equals(sourceColumn.getSourceColumnEnabledReason())) {
                    return;
                }
                // 10.2.还原为初始采集状态
                sourceColumn.setSourceColumnEnabled(true);
                sourceColumn.setSourceColumnEnabledReason(SourceEnabledReason.AUTO);
                sourceColumnService.save(sourceColumn);
            });
        });
    }
    
    @DS("doris")
    @DSTransactional
    public void dealTableAsyncTask(SourceCatalog sourceCatalog,
                                   SourceDatabase sourceDatabase,
                                   Set<String> errorTables,
                                   Set<String> todoAnalyzeTables,
                                   Map<String, List<String>> todoAnalyzeColumnMap) {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        DateTime intervalDay = DateUtil.endOfDay(DateUtil.offsetDay(DateUtil.today(), -sourceProperties.getExpireIntervalDay() - 1), false);
        
        boolean firstTaskFlag = true;
        int totalNum = 0;
        int failedNum = 0;
        int finishedNum = 0;
        
        boolean analyzeDatabaseFinish = false;
        while (!analyzeDatabaseFinish) {
            try {
                // 4.1.等待,优先等待是为了让Doris能够完整的生成任务列表
                Thread.sleep(sourceProperties.getDorisAnalyze().getWaitTime());
                
                // 4.2.查看统计信息收集作业 SHOW ANALYZE
                List<DorisShowAnalyzeTask> tasks = sourceTableDorisAnalyzeService.showAnalyze()
                                                                                 .stream()
                                                                                 .filter(task -> StringUtils.equals(task.getCatalogName(), dorisCatalogName) && StringUtils.equals(task.getDatabaseName(), databaseName))
                                                                                 .toList();
                // 统计执行进度
                if (firstTaskFlag) {
                    totalNum = tasks.size();
                    firstTaskFlag = false;
                }
                failedNum = failedNum + tasks.stream()
                                             .filter(task -> StringUtils.equalsAny(task.getState(), "FAILED"))
                                             .toList()
                                             .size();
                finishedNum = finishedNum + tasks.stream()
                                                 .filter(task -> StringUtils.equalsAny(task.getState(), "FINISHED"))
                                                 .toList()
                                                 .size();
                log.info("[SYNC] {} -> {} -> Doris统计 -> 处理表信息 => 进度 {} / {}", catalogName, databaseName, String.format("%4d", failedNum + finishedNum), String.format("%4d", totalNum));
                
                analyzeDatabaseFinish = true;
                for (DorisShowAnalyzeTask task : tasks) {
                    final String tableName = task.getTableName();
                    switch (task.getState()) {
                        // 4.3.对于 STATE = 'RUNNING' 和 'PENDING' -> 继续循环
                        case "RUNNING", "PENDING" -> analyzeDatabaseFinish = false;
                        // 4.4.对于 STATE = 'FAILED'
                        case "FAILED" -> {
                            log.debug("[SYNC] {} -> {} -> Doris统计 -> 处理表信息 -> {} => [失败] {}", catalogName, databaseName, tableName, task.getMessage());
                            
                            // SHOW TABLE STATS
                            SourceTableDorisAnalyze sourceTableDorisAnalyze = sourceTableDorisAnalyzeService.showTableStats(sourceCatalog, sourceDatabase, tableName);
                            
                            // 提取表统计数据
                            if (intervalDay.before(sourceTableDorisAnalyze.getUpdatedTime())) {
                                SourceTable sourceTable = sourceTableService.get(new SourceTableKey(catalogName, databaseName, tableName));
                                sourceTable.sourceTableDorisAnalyze(sourceTableDorisAnalyze);
                                sourceTableService.save(sourceTable);
                            }
                            
                            // DROP ANALYZE JOB
                            sourceTableDorisAnalyzeService.dropAnalyzeJob(task.getJobId());
                            // 记录失败的表
                            errorTables.add(tableName);
                            // 待分析表
                            todoAnalyzeTables.remove(tableName);
                        }
                        // 4.5.对于 STATE = 'FINISHED'
                        case "FINISHED" -> {
                            // SHOW TABLE STATS
                            SourceTableDorisAnalyze sourceTableDorisAnalyze = sourceTableDorisAnalyzeService.showTableStats(sourceCatalog, sourceDatabase, tableName);
                            
                            // 保存分析记录
                            sourceTableDorisAnalyzeService.save(sourceTableDorisAnalyze);
                            
                            // 提取表统计数据
                            SourceTable sourceTable = sourceTableService.get(new SourceTableKey(catalogName, databaseName, tableName));
                            sourceTable.sourceTableDorisAnalyze(sourceTableDorisAnalyze);
                            sourceTableService.save(sourceTable);
                            
                            // SHOW COLUMN STATS
                            List<SourceColumnDorisAnalyze> sourceColumnDorisAnalyzes = sourceTableDorisAnalyzeService.showColumnsStats(sourceCatalog, sourceDatabase, tableName);
                            // 保存分析记录
                            sourceColumnDorisAnalyzeService.saveList(sourceColumnDorisAnalyzes);
                            
                            // 提取列统计数据
                            List<SourceColumn> sourceColumns = sourceColumnService.listByTable(catalogName, databaseName, tableName);
                            Map<String, SourceColumn> sourceColumnMap = sourceColumns.stream()
                                                                                     .collect(Collectors.toMap(SourceColumn::getColumnName, Function.identity()));
                            for (SourceColumnDorisAnalyze sourceColumnDorisAnalyze : sourceColumnDorisAnalyzes) {
                                SourceColumn sourceColumn = sourceColumnMap.get(sourceColumnDorisAnalyze.getColumnName());
                                sourceColumn.sourceColumnDorisAnalyze(sourceColumnDorisAnalyze);
                                sourceColumnService.save(sourceColumn);
                                
                                // 提取未分析列
                                todoAnalyzeColumnMap.get(sourceColumn.getTableName())
                                                    .remove(sourceColumn.getColumnName());
                            }
                            
                            // DROP ANALYZE JOB
                            sourceTableDorisAnalyzeService.dropAnalyzeJob(task.getJobId());
                            
                            // 待分析表
                            todoAnalyzeTables.remove(tableName);
                        }
                    }
                }
            } catch (InterruptedException e) {
                log.info("[SYNC] {} -> {} -> Doris统计 -> 处理表信息 => [异常] => {}", catalogName, databaseName, e.getMessage());
                analyzeDatabaseFinish = true;
            }
        }
        
        // 待分析列处理
        todoAnalyzeColumnMap.values().removeIf(List::isEmpty);
        
        stopWatch.stop();
        
        log.info("[SYNC] {} -> {} -> Doris统计 -> 处理表信息 => 耗时 {} s", catalogName, databaseName, String.format("%4.0f", stopWatch.getTotalTimeSeconds()));
    }
    
    @DS("doris")
    @DSTransactional
    public void dealColumnAsyncTask(SourceCatalog sourceCatalog,
                                    SourceDatabase sourceDatabase,
                                    Map<String, List<String>> todoAnalyzeColumnMap) {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        boolean firstTaskFlag = true;
        int totalNum = 0;
        int failedNum = 0;
        int finishedNum = 0;
        
        boolean analyzeColumnFinish = false;
        while (!analyzeColumnFinish) {
            try {
                // 7.1.等待,优先等待是为了让Doris能够完整的生成任务列表
                Thread.sleep(sourceProperties.getDorisAnalyze().getWaitTime());
                
                // 7.2.查看统计信息收集作业 SHOW ANALYZE
                List<DorisShowAnalyzeTask> tasks = sourceTableDorisAnalyzeService.showAnalyze()
                                                                                 .stream()
                                                                                 .filter(task -> StringUtils.equals(dorisCatalogName, task.getCatalogName()) && StringUtils.equals(task.getDatabaseName(), databaseName))
                                                                                 .toList();
                // 统计执行进度
                if (firstTaskFlag) {
                    totalNum = tasks.size();
                    firstTaskFlag = false;
                }
                failedNum = failedNum + tasks.stream()
                                             .filter(task -> StringUtils.equalsAny(task.getState(), "FAILED"))
                                             .toList()
                                             .size();
                finishedNum = finishedNum + tasks.stream()
                                                 .filter(task -> StringUtils.equalsAny(task.getState(), "FINISHED"))
                                                 .toList()
                                                 .size();
                log.info("[SYNC] {} -> {} -> Doris统计 -> 处理列信息 => 进度 {} / {}", catalogName, databaseName, String.format("%4d", failedNum + finishedNum), String.format("%4d", totalNum));
                
                analyzeColumnFinish = true;
                for (DorisShowAnalyzeTask task : tasks) {
                    final String tableName = task.getTableName();
                    final String columnName = task.getSingleColumnName(tableName);
                    switch (task.getState()) {
                        // 7.3.对于 STATE = 'RUNNING' 和 'PENDING' -> 继续循环
                        case "RUNNING", "PENDING" -> analyzeColumnFinish = false;
                        
                        // 7.4.对于 STATE = 'FAILED'   -> DROP ANALYZE JOB -> 记录失败的[表:列]Map
                        case "FAILED" -> {
                            // DROP ANALYZE JOB
                            sourceTableDorisAnalyzeService.dropAnalyzeJob(task.getJobId());
                        }
                        // 7.5.对于 STATE = 'FINISHED' -> -> DROP ANALYZE JOB -> SHOW COLUMN STATS 提取列统计数据
                        case "FINISHED" -> {
                            // SHOW COLUMN STATS
                            SourceColumnDorisAnalyze sourceColumnDorisAnalyze = sourceTableDorisAnalyzeService.showColumnStats(sourceCatalog, sourceDatabase, tableName, columnName);
                            
                            // 保存分析记录
                            sourceColumnDorisAnalyzeService.save(sourceColumnDorisAnalyze);
                            
                            // 提取列统计数据
                            SourceColumn sourceColumn = sourceColumnService.get(new SourceColumnKey(catalogName, databaseName, tableName, columnName));
                            sourceColumn.sourceColumnDorisAnalyze(sourceColumnDorisAnalyze);
                            sourceColumnService.save(sourceColumn);
                            
                            // DROP ANALYZE JOB
                            sourceTableDorisAnalyzeService.dropAnalyzeJob(task.getJobId());
                            
                            // 待分析的列
                            todoAnalyzeColumnMap.get(tableName).remove(columnName);
                        }
                    }
                }
            } catch (InterruptedException e) {
                log.info("[SYNC] {} -> {} -> Doris统计 -> 处理列信息 => [异常] => {}", catalogName, databaseName, e.getMessage());
                analyzeColumnFinish = true;
            }
        }
        
        // 待分析列处理
        todoAnalyzeColumnMap.values().removeIf(List::isEmpty);
        
        stopWatch.stop();
        
        log.info("[SYNC] {} -> {} -> Doris统计 -> 处理列信息 => 耗时 {} s", catalogName, databaseName, String.format("%4.0f", stopWatch.getTotalTimeSeconds()));
    }
    
    @DS("doris")
    @DSTransactional
    public List<SourceColumnDorisAnalyze> showColumnsStats(SourceCatalog sourceCatalog,
                                                           SourceDatabase sourceDatabase,
                                                           String tableName) {
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        List<SourceColumnDorisAnalyze> sourceColumnDorisAnalyzes = new ArrayList<>();
        String sql = String.format("SHOW COLUMN STATS `%s`.`%s`.`%s`", dorisCatalogName, databaseName, tableName);
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            ResultSet rs = connection.createStatement().executeQuery(sql);
            while (rs.next()) {
                SourceColumnDorisAnalyze sourceColumnDorisAnalyze = new SourceColumnDorisAnalyze();
                sourceColumnDorisAnalyze.setCatalogName(catalogName);
                sourceColumnDorisAnalyze.setDatabaseName(databaseName);
                sourceColumnDorisAnalyze.setTableName(tableName);
                sourceColumnDorisAnalyze.setColumnName(rs.getString("column_name"));
                sourceColumnDorisAnalyze.setUpdatedTime(DateUtil.parse(rs.getString("updated_time")));
                sourceColumnDorisAnalyze.setIndexName(rs.getString("index_name"));
                sourceColumnDorisAnalyze.setCount(rs.getLong("count"));
                sourceColumnDorisAnalyze.setNdv(rs.getLong("ndv"));
                sourceColumnDorisAnalyze.setNumNull(rs.getLong("num_null"));
                sourceColumnDorisAnalyze.setDataSize(rs.getLong("data_size"));
                sourceColumnDorisAnalyze.setAvgSizeByte(rs.getDouble("avg_size_byte"));
                sourceColumnDorisAnalyze.setMin(rs.getString("min"));
                sourceColumnDorisAnalyze.setMax(rs.getString("max"));
                sourceColumnDorisAnalyze.setMethod(rs.getString("method"));
                sourceColumnDorisAnalyze.setType(rs.getString("type"));
                sourceColumnDorisAnalyze.setTrigger(rs.getString("trigger"));
                sourceColumnDorisAnalyze.setQueryTimes(rs.getInt("query_times"));
                
                sourceColumnDorisAnalyzes.add(sourceColumnDorisAnalyze);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return sourceColumnDorisAnalyzes;
    }
    
    @DS("doris")
    @DSTransactional
    public SourceColumnDorisAnalyze showColumnStats(SourceCatalog sourceCatalog,
                                                    SourceDatabase sourceDatabase,
                                                    String tableName,
                                                    String columnName) {
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        SourceColumnDorisAnalyze sourceColumnDorisAnalyze = null;
        String sql = String.format("SHOW COLUMN STATS `%s`.`%s`.`%s`(`%s`)", dorisCatalogName, databaseName, tableName, columnName);
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            ResultSet rs = connection.createStatement().executeQuery(sql);
            if (rs.next()) {
                sourceColumnDorisAnalyze = new SourceColumnDorisAnalyze();
                sourceColumnDorisAnalyze.setCatalogName(catalogName);
                sourceColumnDorisAnalyze.setDatabaseName(databaseName);
                sourceColumnDorisAnalyze.setTableName(tableName);
                sourceColumnDorisAnalyze.setColumnName(rs.getString("column_name"));
                sourceColumnDorisAnalyze.setUpdatedTime(DateUtil.parse(rs.getString("updated_time")));
                sourceColumnDorisAnalyze.setIndexName(rs.getString("index_name"));
                sourceColumnDorisAnalyze.setCount(rs.getLong("count"));
                sourceColumnDorisAnalyze.setNdv(rs.getLong("ndv"));
                sourceColumnDorisAnalyze.setNumNull(rs.getLong("num_null"));
                sourceColumnDorisAnalyze.setDataSize(rs.getLong("data_size"));
                sourceColumnDorisAnalyze.setAvgSizeByte(rs.getDouble("avg_size_byte"));
                sourceColumnDorisAnalyze.setMin(rs.getString("min"));
                sourceColumnDorisAnalyze.setMax(rs.getString("max"));
                sourceColumnDorisAnalyze.setMethod(rs.getString("method"));
                sourceColumnDorisAnalyze.setType(rs.getString("type"));
                sourceColumnDorisAnalyze.setTrigger(rs.getString("trigger"));
                sourceColumnDorisAnalyze.setQueryTimes(rs.getInt("query_times"));
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return sourceColumnDorisAnalyze;
    }
    
    @DS("doris")
    @DSTransactional
    public SourceTableDorisAnalyze showTableStats(SourceCatalog sourceCatalog,
                                                  SourceDatabase sourceDatabase,
                                                  String tableName) {
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        SourceTableDorisAnalyze sourceTableDorisAnalyze = null;
        String sql = String.format("SHOW TABLE STATS `%s`.`%s`.`%s`", dorisCatalogName, databaseName, tableName);
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            ResultSet rs = connection.createStatement().executeQuery(sql);
            if (rs.next()) {
                sourceTableDorisAnalyze = new SourceTableDorisAnalyze();
                sourceTableDorisAnalyze.setCatalogName(catalogName);
                sourceTableDorisAnalyze.setDatabaseName(databaseName);
                sourceTableDorisAnalyze.setTableName(tableName);
                sourceTableDorisAnalyze.setUpdatedTime(rs.getTimestamp("updated_time"));
                sourceTableDorisAnalyze.setUpdatedRows(rs.getLong("updated_rows"));
                sourceTableDorisAnalyze.setQueryTimes(rs.getInt("query_times"));
                sourceTableDorisAnalyze.setRowCount(rs.getLong("row_count"));
                sourceTableDorisAnalyze.setColumns(rs.getString("columns"));
                sourceTableDorisAnalyze.setTrigger(rs.getString("trigger"));
                sourceTableDorisAnalyze.setNewPartition(rs.getBoolean("new_partition"));
                sourceTableDorisAnalyze.setUserInject(rs.getBoolean("user_inject"));
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return sourceTableDorisAnalyze;
    }
    
    @DS("doris")
    @DSTransactional
    public List<DorisShowAnalyzeTask> showAnalyze() {
        List<DorisShowAnalyzeTask> dorisShowAnalyzeTasks = new ArrayList<>();
        String sql = "SHOW ANALYZE";
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            ResultSet rs = connection.createStatement().executeQuery(sql);
            while (rs.next()) {
                DorisShowAnalyzeTask dorisShowAnalyzeTask = new DorisShowAnalyzeTask();
                dorisShowAnalyzeTask.setJobId(rs.getLong("job_id"));
                dorisShowAnalyzeTask.setCatalogName(rs.getString("catalog_name"));
                dorisShowAnalyzeTask.setDatabaseName(rs.getString("db_name"));
                dorisShowAnalyzeTask.setTableName(rs.getString("tbl_name"));
                dorisShowAnalyzeTask.setColumnName(rs.getString("col_name"));
                dorisShowAnalyzeTask.setJobType(rs.getString("job_type"));
                dorisShowAnalyzeTask.setAnalysisType(rs.getString("analysis_type"));
                dorisShowAnalyzeTask.setMessage(rs.getString("message"));
                dorisShowAnalyzeTask.setLastExecTimeInMs(rs.getTimestamp("last_exec_time_in_ms"));
                dorisShowAnalyzeTask.setState(rs.getString("state"));
                dorisShowAnalyzeTask.setProgress(rs.getString("progress"));
                dorisShowAnalyzeTask.setScheduleType(rs.getString("schedule_type"));
                dorisShowAnalyzeTask.setStartTime(rs.getTimestamp("start_time"));
                dorisShowAnalyzeTask.setEndTime(rs.getTimestamp("end_time"));
                
                dorisShowAnalyzeTasks.add(dorisShowAnalyzeTask);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return dorisShowAnalyzeTasks;
    }
    
    @DS("doris")
    @DSTransactional
    public void clearAnalyzeJob(SourceCatalog sourceCatalog, SourceDatabase sourceDatabase) {
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String catalogName = sourceCatalog.getCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        // 1.1.查看统计信息收集作业 SHOW ANALYZE
        List<DorisShowAnalyzeTask> tasks = sourceTableDorisAnalyzeService.showAnalyze();
        if (!tasks.isEmpty()) {
            tasks = tasks.stream()
                         .filter(task -> StringUtils.equals(dorisCatalogName, task.getCatalogName()) && StringUtils.equals(databaseName, task.getDatabaseName()))
                         .toList();
            if (!tasks.isEmpty()) {
                for (DorisShowAnalyzeTask task : tasks) {
                    switch (task.getState()) {
                        case "RUNNING", "PENDING" -> {
                            // 1.2.停止正在后台执行的统计信息收集作业 KILL ANALYZE
                            sourceTableDorisAnalyzeService.killAnalyzeJob(task.getJobId());
                        }
                        case "FINISHED", "FAILED" -> {
                            // 1.3 删除指定的统计信息收集作业的历史记录 DROP ANALYZE
                            sourceTableDorisAnalyzeService.dropAnalyzeJob(task.getJobId());
                        }
                    }
                }
            }
        }
    }
    
    @DS("doris")
    @DSTransactional
    public void killAnalyzeJob(Long jobId) {
        String sql = "KILL ANALYZE " + jobId;
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            connection.createStatement().execute(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    @DS("doris")
    @DSTransactional
    public void dropAnalyzeJob(Long jobId) {
        String sql = "DROP ANALYZE JOB " + jobId;
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            connection.createStatement().execute(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    @DS("doris")
    @DSTransactional
    public void analyzeDatabase(SourceCatalog sourceCatalog, SourceDatabase sourceDatabase) {
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        String sql = String.format("ANALYZE DATABASE `%s`.`%s`", dorisCatalogName, databaseName);
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            connection.createStatement().execute(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    @DS("doris")
    @DSTransactional
    public void analyzeTable(SourceCatalog sourceCatalog, SourceDatabase sourceDatabase, String tableName) {
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        String sql = String.format("ANALYZE TABLE `%s`.`%s`.`%s`", dorisCatalogName, databaseName, tableName);
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            connection.createStatement().execute(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    @DS("doris")
    @DSTransactional
    public void analyzeColumn(SourceCatalog sourceCatalog,
                              SourceDatabase sourceDatabase,
                              String tableName,
                              String columnName) {
        String catalogName = sourceCatalog.getCatalogName();
        String dorisCatalogName = sourceCatalog.getDorisCatalogName();
        String databaseName = sourceDatabase.getDatabaseName();
        
        // todo 过滤一下诸如 BLOB CLOB 类型的列, 不进行分析
        switch (sourceCatalog.getDbType()) {
            case ORACLE_SERVICE, ORACLE_SID -> {
                SourceColumn sourceColumn = sourceColumnService.get(new SourceColumnKey(catalogName, databaseName, tableName, columnName));
                String dataType = sourceColumn.getDataType();
                if (StringUtils.equalsAny(dataType, "BLOB", "CLOB")) {
                    log.info("[SYNC] {} -> {} -> Doris统计 -> 分析列     => {}.{} => [取消] 数据类型 {}", catalogName, databaseName, tableName, columnName, dataType);
                    return;
                }
            }
        }
        
        String sql = String.format("ANALYZE TABLE `%s`.`%s`.`%s`(`%s`)", dorisCatalogName, databaseName, tableName, columnName);
        try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {
            connection.createStatement().execute(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}